package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.{Before, Test}

class Demo16Student {
  var students: RDD[Students] = _
  var scores: RDD[Score] = _
  var subject: RDD[Subject] = _
  var conf: SparkConf = _
  var sc: SparkContext = _


  @Before
  def read_file(): Unit = {
    // 构建Spark上下文环境
    conf = new SparkConf()
    conf.setAppName("Demo12MapValues")
    conf.setMaster("local[*]")
    sc = new SparkContext(conf)

    // 读取students数据并构建RDD，以Students样例类对象作为RDD中的每个元素
    students = sc.textFile("data/stu/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Students(splits(0).toInt, splits(1), splits(2).toInt, splits(3), splits(4))
      })



    // 读取score数据构建RDD，以Score样例类对象作为RDD中的每个元素

    scores = sc.textFile("data/stu/score.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Score(splits(0).toInt, splits(1).toInt, splits(2).toInt)
      })

    // 读取subject数据构建RDD，以Subject样例类对象作为RDD中的每个元素

    subject = sc.textFile("data/stu/subject.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Subject(splits(0).toInt, splits(1), splits(2).toInt)
      })


  }

  @Test
  // 5、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数]
  def question5(): Unit = {
    // 归一化
    // 将所有的分数转换成百分制

    val sid_score_KVRDD: RDD[(Int, Int)] = subject.map(sub => (sub.subject_id, sub.subject_score))

    val sid_sco_KVRDD: RDD[(Int, Score)] = scores.map(sco => (sco.subject_id, sco))

    //    val new_scores: RDD[(Int, Int, Double)] = scores
    //            .map(sco => {
    //              val sub_score: Int = sid_score_map(sco.subject_id)
    //              (sco.id, sco.subject_id, sco.score / sub_score.toDouble * 100)
    //            })
    val id_sid_new_scoreRDD: RDD[(Int, Int, Double)] = sid_sco_KVRDD.join(sid_score_KVRDD).map {
      case (sid: Int, (sco: Score, full_score: Int)) =>
        (sco.id, sco.subject_id, sco.score / full_score.toDouble * 100)
    }


    // 计算方差
    // 偏科最严重的前100名学生id
    id_sid_new_scoreRDD
      .groupBy(_._1)
      .map {
        case (id: Int, scoreIter: Iterable[(Int, Int, Double)]) =>
          // 计算平均分
          val avg_score: Double = scoreIter.map(_._3).sum / scoreIter.size
          // 计算方差
          // 计算 Sum(x-avg_score)^2 / N
          val variance: Double = scoreIter.map {
            case (id: Int, sid: Int, new_score: Double) =>
              Math.pow(new_score - avg_score, 2)
          }.sum / scoreIter.size
          (id, variance)
      }.sortBy(-_._2)
      .take(100)
      .foreach(println)
//      .map(_._1)


  }

}

case class Students(id: Int, name: String, age: Int, gender: String, clazz: String)

case class Score(id: Int, subject_id: Int, score: Int)

case class Subject(subject_id: Int, subject_name: String, subject_score: Int)
